package com.lagou.niancheng.work2.filter;

import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.springframework.util.CollectionUtils;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;

/**
 * com.lagou.niancheng.work2.filter.TPMonitorFilter
 *
 * @author 大犬
 * @date 2020/6/8
 */
public class TPMonitorFilter implements Filter {

    private static final Map<String, List<MethodResponseMetaData>> METHOD_RESPONSE_TIME_CACHE1 = new ConcurrentHashMap<>();

    private static final Map<String, List<MethodResponseMetaData>[]> METHOD_RESPONSE_TIME_CACHE2 = new ConcurrentHashMap<>();

    private static final AtomicLong INDEXER = new AtomicLong(0);

    private static final int CACHE_SIZE = 12;

    public TPMonitorFilter() {
        // 开启定时5s一次计算TP90和TP99
        Executors.newSingleThreadScheduledExecutor()
                .scheduleAtFixedRate(this::doRefresh, 5, 5, TimeUnit.SECONDS);
    }


    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String methodName = invocation.getMethodName();
        return recordResponseTime(methodName, () -> invoker.invoke(invocation));
    }


    private <T> T recordResponseTime(String methodName, Supplier<T> supplier) {
        long startTime = System.currentTimeMillis();
        try {
            return supplier.get();
        } finally {
            // 记录响应耗时
            long responseTime = System.currentTimeMillis() - startTime;
            MethodResponseMetaData methodResponseMetaData = new MethodResponseMetaData();
            methodResponseMetaData.setRefreshIndex(INDEXER.get() % CACHE_SIZE);
            methodResponseMetaData.setResponseTime(responseTime);
            doRecordResponseTime(methodName, methodResponseMetaData);
        }
    }

    private void doRecordResponseTime(String methodName, MethodResponseMetaData responseTimeMetaData) {
        synchronized (methodName.intern()) {
            // 将记录存入一级缓存中
            METHOD_RESPONSE_TIME_CACHE1.putIfAbsent(methodName, new LinkedList<>());
            List<MethodResponseMetaData> methodResponseMetaDatas = METHOD_RESPONSE_TIME_CACHE1.get(methodName);
            methodResponseMetaDatas.add(responseTimeMetaData);
        }
    }

    @SuppressWarnings("unchecked")
    private void doRefresh() {
        // 刷新索引+1
        int currentIndex = (int) (INDEXER.getAndIncrement() % CACHE_SIZE);

        // 1.将一级缓存中的数据放入二级缓存中
        METHOD_RESPONSE_TIME_CACHE1.forEach((methodName, list) -> {
            synchronized (methodName.intern()) {
                // 去除一分钟之外的数据，添加新的数据
                METHOD_RESPONSE_TIME_CACHE2.putIfAbsent(methodName, new List[CACHE_SIZE]);
                List<MethodResponseMetaData>[] zoneDataArray = METHOD_RESPONSE_TIME_CACHE2.get(methodName);
                zoneDataArray[currentIndex] = list;
                // 清空一级缓存
                METHOD_RESPONSE_TIME_CACHE1.remove(methodName);
            }
        });

        // 2.遍历二级缓存，按响应时间排序找出tp90和tp99
        System.out.println("======================" + currentIndex + "==========================");
        METHOD_RESPONSE_TIME_CACHE2.forEach((methodName, zoneDataArray) -> {
            List<Long> allResponseDatas = new ArrayList<>();
            for (List<MethodResponseMetaData> methodResponseMetaDataList : zoneDataArray) {
                if (CollectionUtils.isEmpty(methodResponseMetaDataList)) {
                    continue;
                }
                methodResponseMetaDataList.stream()
                        .map(MethodResponseMetaData::getResponseTime)
                        .forEach(allResponseDatas::add);
            }
            // 按响应事件排序
            allResponseDatas.sort(Comparator.comparingLong(Long::longValue));
            int size = allResponseDatas.size();
            int top90Idx = size - (int) Math.ceil(size * 0.9);
            int top99Idx = size - (int) Math.ceil(size * 0.99);
            System.out.println(MessageFormat.format("方法{0}一分钟内有{1}次调用 他的tp90值为：{2} ; tp99值为：{3}",
                    methodName, size, allResponseDatas.get(top90Idx), allResponseDatas.get(top99Idx)));
        });
        System.out.println();
    }

}
